Spark Tuning - Performance Optimizations: Hands-on Cache Workbook
This workbook walks you through configuring and validating dynamic cache persistency.
1. Caching Configuration
- Task: Ingest a dataset, execute a multi-step filter, and cache intermediate results using explicit storage levels.
2. Tasks
Task 1: Configure Persist storage levels
Write the PySpark DataFrame code to load a Parquet dataset, filter high-value orders, cache it using MEMORY_AND_DISK storage serialization level, and verify its storage status in the Spark UI.
Task 2: Validate AQE Coalesce Decisions
Trace the executor console logs when a query with 200 partition shuffles is coalesced by AQE down to 8 partitions. Write down the config settings required to enable AQE and specify minimum coalesced partition sizes.
3. Step-by-Step Solutions
Solution 1: PySpark implementation
- PySpark Code:
from pyspark import StorageLevel
# 1. Read dataset
orders_df = spark.read.parquet("hdfs://cluster/orders")
# 2. Filter records
large_orders = orders_df.filter("amount > 1000")
# 3. Apply explicit Storage Level
large_orders.persist(StorageLevel.MEMORY_AND_DISK)
# 4. Trigger cache allocation via a count action
cached_count = large_orders.count()
print(f"Cached records count: {cached_count}")
# 5. Inspect execution
# Go to Spark UI (port 4040) -> 'Storage' tab to verify the memory footprint.
# 6. Unpersist memory once ETL completes
large_orders.unpersist()
Solution 2: Enabling AQE
- Configuration Settings:
# Enable Adaptive Query Execution
spark.conf.set("spark.sql.adaptive.enabled", "true")
# Set target minimum size for coalesced partitions (default: 64MB)
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "67108864")
# Configure minimum number of partitions after coalescing
spark.conf.set("spark.sql.adaptive.coalescePartitions.minPartitionNum", "1")
Find this content helpful?
☕ Buy me a coffee